package com.chencong.online.function;

import com.chencong.online.bean.PageViewCountBean;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @program: user-behavior-analysis-online
 * @ClassName PageViewCountProcessFunc
 * @description:按照窗口分组统计每个窗口值
 * @author: chencong
 * @create: 2021-12-28 11:55
 **/
public class PageViewCountProcessFunc extends KeyedProcessFunction<Long, PageViewCountBean, PageViewCountBean> {
    //定义状态
    private ValueState<Long> totalValueCount;
    private Long windowStart;

    @Override
    public void open(Configuration parameters) throws Exception {
        totalValueCount = getRuntimeContext().getState(new ValueStateDescriptor<Long>("total-count-state", Long.class,0L));
    }

    @Override
    public void processElement(PageViewCountBean value, KeyedProcessFunction<Long, PageViewCountBean, PageViewCountBean>.Context ctx, Collector<PageViewCountBean> out) throws Exception {
        //将上一次状态取出后加入现有值后更新到最新状态
        totalValueCount.update(totalValueCount.value() + value.getCount());
        //注册定时器
        ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 1);
        //获取时间开始窗口
        windowStart = value.getWindowStart();

    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<Long, PageViewCountBean, PageViewCountBean>.OnTimerContext ctx, Collector<PageViewCountBean> out) throws Exception {
        // 定时器触发，所有分组count值都到齐，直接输出当前的总count数量
        out.collect(new PageViewCountBean(windowStart, ctx.getCurrentKey(), "pv", totalValueCount.value()));
        //清空状态
        totalValueCount.clear();
    }

}
